package net.sureon.common.mail;

import java.util.ArrayList;
import java.util.Collection;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * This class holds a number of threads that is used my the Mail Processor
 * to send mail asynchronously.
 */
public class MailThreadPool {
	private static final Log LOGGER = LogFactory.getLog(MailThreadPool.class);
      private Collection units = new ArrayList(3);

      /**
       * mailThreads
       */
      protected Thread[] mailThreads = null;

      /**
       * done
       */
      protected ProcessDone done = new ProcessDone();

      /**
       * Constructor, this sets up the pool size
       *
       * @param size int
       */
      public MailThreadPool(int size) {
        if (size < 1) {
          size = 1; // default
        }
        mailThreads = new MailWorkerThread[size];
        for (int i = 0; i < mailThreads.length; i++) {
          mailThreads[i] = new MailWorkerThread(this);
          mailThreads[i].start();
        }
      }

      /**
       * Assigns the runnable unit to the queue for the thread pool to pick up and execute
       *
       * @param runnableUnit Runnable
       */
      public synchronized void assignUnit(Runnable runnableUnit) {
          if(LOGGER.isInfoEnabled()) {
              //if(runnableUnit instanceof MailProcessorImpl.MailProcessingUnit){
              //   MailProcessorImpl.MailProcessingUnit unit = (MailProcessorImpl.MailProcessingUnit)runnableUnit;
              //    logger.logInfo("Assign Process Unit:" + unit.getKey().toString());
              //}

          }
        done.workerBegin();
        units.add(runnableUnit);
        notify();
      }

      /**
       * Gets the next runnable unit from the queue.
       *
       * @return java.lang.Runnable
       */
      public synchronized Runnable getUnit() {
        try {
          while (!units.iterator().hasNext()) {
            wait();
          }

          Runnable r = (Runnable) units.iterator().next();
          units.remove(r);
          return r;
        } catch (InterruptedException e) {
          done.workerEnd();
          return null;
        }
      }

      /**
       * Called when all the thread pools are no longer needed
       */
      public void complete() {
        done.waitBegin();
        done.waitDone();
      }

      /**
       * finalize
       */
      protected void finalize() {
        done.reset();
        for (int i = 0; i < mailThreads.length; i++) {
          mailThreads[i].interrupt();
          done.workerBegin();
          mailThreads[i].destroy();
        }
        done.waitDone();
      }

      /**
       * Inner class that are the "Threads" in the pool. It is these
       * threads that execute the runnable units
       */
      class MailWorkerThread
          extends Thread {
        //private boolean busy;
        private MailThreadPool owner;

        MailWorkerThread(MailThreadPool o) {
          owner = o;
        }
        /**
         * run
         */
        public void run() {
          Runnable target = null;

          do {
            target = owner.getUnit();
            if (target != null) {
              target.run();
              owner.done.workerEnd();
            }
          }
          while (target != null);
        }
      }

      /**
       * Inner class that keeps tracks the thread processes
       */
      class ProcessDone {

        private int activeThreads = 0;

        private boolean started = false;

        synchronized void waitDone() {
          try {
            while (activeThreads > 0) {
              wait();
            }
          } catch (InterruptedException e) {
            LOGGER.error(e);
          }
        }

        synchronized void waitBegin() {
          try {
            while (!started) {
              wait();
            }
          } catch (InterruptedException e) {
              LOGGER.error(e);
          }
        }

        synchronized void workerBegin() {
          activeThreads++;
          started = true;
          notify();
        }

        synchronized void workerEnd() {
          activeThreads--;
          notify();
        }

        synchronized void reset() {
          activeThreads = 0;
        }

      }

}
